package com.fourth.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fourth.bean.InteractionReviewWindow;
import com.fourth.utils.DateFormatUtil;
import com.fourth.utils.MyClickHouseUtil;
import com.fourth.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author yangyu
 * @create 2022/8/20
 */
public class DwsInteractionReviewWindow {

    public static void main(String[] args) throws Exception {

        //1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取kafka dwd层课程评价
        String topic = "dwd_interaction_review";
        String groupId = "review_info_22";
        DataStreamSource<String> kafkaDS  = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //3、将数据流转换为JavaBean
        SingleOutputStreamOperator<InteractionReviewWindow> interReviewDS = kafkaDS.map(line -> {

            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject data = jsonObject.getJSONObject("data");

            return InteractionReviewWindow.builder()
                    .courseId(data.getString("course_id"))
                    .userID(1L)
                    .reviewStars(data.getDouble("review_stars"))
                    .ts(DateFormatUtil.toTs(data.getString("create_time"), true))
                    .build();
        });

        //4、分组开窗聚合
        SingleOutputStreamOperator<InteractionReviewWindow> reduceDS = interReviewDS.assignTimestampsAndWatermarks(WatermarkStrategy.<InteractionReviewWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<InteractionReviewWindow>() {
            @Override
            public long extractTimestamp(InteractionReviewWindow element, long recordTimestamp) {
                return element.getTs();
            }
        }))
                .keyBy(InteractionReviewWindow::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<InteractionReviewWindow>() {
                    @Override
                    public InteractionReviewWindow reduce(InteractionReviewWindow value1, InteractionReviewWindow value2) throws Exception {
                        value1.setUserID(value1.getUserID() + value2.getUserID());
                        value1.setReviewStars(value1.getReviewStars() + value2.getReviewStars());
                        value1.setPositiveRatio(value1.getReviewStars()/value1.getUserID());
                        return value1;
                    }
                }, new WindowFunction<InteractionReviewWindow, InteractionReviewWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<InteractionReviewWindow> input, Collector<InteractionReviewWindow> out) throws Exception {

                        InteractionReviewWindow next = input.iterator().next();

                        next.setTs(System.currentTimeMillis());
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));

                        out.collect(next);
                    }
                });
        reduceDS.print("reduceDS>>>>>>>>>>>>>");

        //5、将数据写到ClickHouse
        reduceDS.print(">>>>>>>>>>>>");
        reduceDS.addSink(MyClickHouseUtil.getSink(
                "insert into dwd_interaction_review values(?,?,?,?,?,?,?)"));

        //8.启动任务
        env.execute("DwsInteractionReviewWindow");

    }
}
